package com.bw.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.MyKafkaUtil;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.text.SimpleDateFormat;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.TimeZone;

public class Test4 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置并行度为1
        env.setParallelism(1);

        DataStreamSource<String> stream = env.addSource(MyKafkaUtil.getKafkaConsumer("tms_ods_yk9","test12"));


        //筛选订单和订单明细数据
        SingleOutputStreamOperator<String> DwdBoundSortBeanDS = stream.filter(new FilterFunction<String>() {
            @Override
            public boolean filter(String s) throws Exception {
                try {
                    JSONObject ob= JSON.parseObject(s);
                    if(ob!=null) {
                        if(ob.getString("table").equals("transport_task")) {
                            String transport_status=ob.getJSONObject("data").getString("status");
                            return transport_status!=null&&transport_status.equals("67004");
                        }
                    }
                }catch (Exception e) {

                }
                return false;
            }
        });
        DwdBoundSortBeanDS.print();

        //
        SingleOutputStreamOperator<String> resultDS=DwdBoundSortBeanDS.map(
                new MapFunction<String, String>() {
                    @Override
                    public String map(String s) throws Exception {

                        //2022-07-11T14:00:27Z


                        JSONObject tt = JSON.parseObject(s).getJSONObject("data");
                        //处理时区问题
                        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss'Z'");
                        sdf.setTimeZone(TimeZone.getTimeZone("GMT+8"));

                        //处理



                        long time = sdf.parse(tt.getString("create_time")).getTime();
                        //补全时间戳字段
                        tt.put("create_ts",time);
                        //补全运输时长字段  毫秒数
                        tt.put("transport_time",tt.getLong("actual_end_time") - tt.getLong("actual_start_time"));
                        //脱敏
                        tt.put("driver1_name",tt.getString("driver1_name") != null ? tt.getString("driver1_name").substring(0, 1) + "**" : tt.getString("driver1_name"));
                        tt.put("truck_no",tt.getString("truck_no") != null ? tt.getString("truck_no").substring(0, 3) + "****" : tt.getString("truck_no"));

                        return JSONObject.toJSONString(tt);
                    }
                }
        );
        resultDS.print();
        resultDS.addSink(MyKafkaUtil.getKafkaProducer("dwd_transport_task_p67004_yk9"));
        env.execute();

    }
}
